Netty 中的粘包和拆包

您所在的位置：网站首页 › netty 粘包解决方案 › Netty 中的粘包和拆包

Netty 中的粘包和拆包

2023-09-21 07:30| 来源: 网络整理| 查看: 265

Netty 底层是基于 TCP 协议来处理网络数据传输。我们知道 TCP 协议是面向字节流的协议，数据像流水一样在网络中传输那何来 “包” 的概念呢？

TCP是四层协议不负责数据逻辑的处理，但是数据在TCP层 “流” 的时候为了保证安全和节约效率会把 “流” 做一些分包处理，比如：

发送方约定了每次数据传输的最大包大小，超过该值的内容将会被拆分成两个包发送；发送端和接收端约定每次发送数据包长度并随着网络状况动态调整接收窗口大小，这里也会出现拆包的情况；

Netty 本身是基于 TCP 协议做的处理，如果它不去对 “流” 进行处理，到底这个 “流” 从哪到哪才是完整的数据就是个迷。我们先来看在 TCP 协议中有哪些步骤可能会让 “流” 不完整或者是出现粘滞的可能。

1. TCP 中可能出现粘包/拆包的原因

数据流在TCP协议下传播，因为协议本身对于流有一些规则的限制，这些规则会导致当前对端接收到的数据包不完整，归结原因有下面三种情况：

Socket 缓冲区与滑动窗口 MSS/MTU限制 Nagle算法 1. Socket缓冲区与滑动窗口

对于 TCP 协议而言，它传输数据是基于字节流传输的。应用层在传输数据时，实际上会先将数据写入到 TCP 套接字的缓冲区，当缓冲区被写满后，数据才会被写出去。每个TCP Socket 在内核中都有一个发送缓冲区(SO_SNDBUF )和一个接收缓冲区(SO_RCVBUF)，TCP 的全双工的工作模式以及 TCP 的滑动窗口便是依赖于这两个独立的 buffer 以及此 buffer 的填充状态。

SO_SNDBUF：

进程发送的数据的时候假设调用了一个 send 方法，将数据拷贝进入 Socket 的内核发送缓冲区之中，然后 send 便会在上层返回。换句话说，send 返回之时，数据不一定会发送到对端去(和write写文件有点类似)，send 仅仅是把应用层 buffer 的数据拷贝进 Socket 的内核发送 buffer 中。

SO_RCVBUF：

把接收到的数据缓存入内核，应用进程一直没有调用 read 进行读取的话，此数据会一直缓存在相应 Socket 的接收缓冲区内。不管进程是否读取 Socket，对端发来的数据都会经由内核接收并且缓存到 Socket 的内核接收缓冲区之中。read 所做的工作，就是把内核缓冲区中的数据拷贝到应用层用户的 buffer 里面，仅此而已。

接收缓冲区保存收到的数据一直到应用进程读走为止。对于 TCP，如果应用进程一直没有读取，buffer 满了之后发生的动作是：通知对端 TCP 协议中的窗口关闭。这个便是滑动窗口的实现。保证 TCP 套接口接收缓冲区不会溢出，从而保证了 TCP 是可靠传输。因为对方不允许发出超过所通告窗口大小的数据。这就是 TCP 的流量控制，如果对方无视窗口大小而发出了超过窗口大小的数据，则接收方 TCP 将丢弃它。

滑动窗口：

TCP连接在三次握手的时候，会将自己的窗口大小(window size)发送给对方，其实就是 SO_RCVBUF 指定的值。之后在发送数据的时，发送方必须要先确认接收方的窗口没有被填充满，如果没有填满，则可以发送。

每次发送数据后，发送方将自己维护的对方的 window size 减小，表示对方的 SO_RCVBUF 可用空间变小。

当接收方处理开始处理 SO_RCVBUF 中的数据时，会将数据从 Socket 在内核中的接受缓冲区读出，此时接收方的 SO_RCVBUF 可用空间变大，即 window size 变大，接受方会以 ack 消息的方式将自己最新的 window size 返回给发送方，此时发送方将自己的维护的接受的方的 window size 设置为ack消息返回的 window size。

此外，发送方可以连续的给接受方发送消息，只要保证对方的 SO_RCVBUF 空间可以缓存数据即可，即 window size>0。当接收方的 SO_RCVBUF 被填充满时，此时 window size=0，发送方不能再继续发送数据，要等待接收方 ack 消息，以获得最新可用的 window size。

2. MSS/MTU分片

MTU (Maxitum Transmission Unit,最大传输单元)是链路层对一次可以发送的最大数据的限制。MSS(Maxitum Segment Size,最大分段大小)是 TCP 报文中 data 部分的最大长度，是传输层对一次可以发送的最大数据的限制。

数据在传输过程中，每经过一层，都会加上一些额外的信息：

应用层：只关心发送的数据 data，将数据写入 Socket 在内核中的缓冲区 SO_SNDBUF 即返回，操作系统会将 SO_SNDBUF 中的数据取出来进行发送；传输层：会在 data 前面加上 TCP Header(20字节)；网络层：会在 TCP 报文的基础上再添加一个 IP Header，也就是将自己的网络地址加入到报文中。IPv4 中 IP Header 长度是 20 字节，IPV6 中 IP Header 长度是 40 字节；链路层：加上 Datalink Header 和 CRC。会将 SMAC(Source Machine，数据发送方的MAC地址)，DMAC(Destination Machine，数据接受方的MAC地址 )和 Type 域加入。SMAC+DMAC+Type+CRC 总长度为 18 字节；物理层：进行传输。

在回顾这个基本内容之后，再来看 MTU 和 MSS。MTU 是以太网传输数据方面的限制，每个以太网帧最大不能超过 1518bytes。刨去以太网帧的帧头(DMAC+SMAC+Type域) 14Bytes 和帧尾 (CRC校验 ) 4 Bytes，那么剩下承载上层协议的地方也就是 data 域最大就只能有 1500 Bytes 这个值我们就把它称之为 MTU。

MSS 是在 MTU 的基础上减去网络层的 IP Header 和传输层的 TCP Header 的部分，这就是 TCP 协议一次可以发送的实际应用数据的最大大小。

MSS = MTU(1500) -IP Header(20 or 40)-TCP Header(20)

由于 IPV4 和 IPV6 的长度不同，在 IPV4 中，以太网 MSS 可以达到 1460byte。在 IPV6 中，以太网 MSS 可以达到 1440byte。

发送方发送数据时，当 SO_SNDBUF 中的数据量大于 MSS 时，操作系统会将数据进行拆分，使得每一部分都小于 MSS，也形成了拆包。然后每一部分都加上 TCP Header，构成多个完整的 TCP 报文进行发送，当然经过网络层和数据链路层的时候，还会分别加上相应的内容。

另外需要注意的是：对于本地回环地址(lookback)不需要走以太网，所以不受到以太网 MTU=1500 的限制。linux 服务器上输入 ifconfig 命令，可以查看不同网卡的 MTU 大小，如下：

上图显示了 2 个网卡信息：

eth0 需要走以太网，所以 MTU 是 1500； lo 是本地回环，不需要走以太网，所以不受 1500 的限制。

Nagle 算法

TCP/IP 协议中，无论发送多少数据，总是要在数据(data)前面加上协议头(TCP Header+IP Header)，同时，对方接收到数据，也需要发送 ACK 表示确认。

即使从键盘输入的一个字符，占用一个字节，可能在传输上造成 41 字节的包，其中包括 1 字节的有用信息和 40 字节的首部数据。这种情况转变成了 4000% 的消耗，这样的情况对于重负载的网络来是无法接受的。称之为"糊涂窗口综合征"。

为了尽可能的利用网络带宽，TCP 总是希望尽可能的发送足够大的数据。(一个连接会设置 MSS 参数，因此，TCP/IP 希望每次都能够以 MSS 尺寸的数据块来发送数据)。Nagle 算法就是为了尽可能发送大块数据，避免网络中充斥着许多小数据块。

Nagle 算法的基本定义是任意时刻，最多只能有一个未被确认的小段。所谓 “小段”，指的是小于 MSS 尺寸的数据块；所谓“未被确认”，是指一个数据块发送出去后，没有收到对方发送的 ACK 确认该数据已收到。

Nagle 算法的规则：

如果 SO_SNDBUF 中的数据长度达到 MSS，则允许发送；如果该 SO_SNDBUF 中含有 FIN，表示请求关闭连接，则先将 SO_SNDBUF 中的剩余数据发送，再关闭；设置了 TCP_NODELAY=true 选项，则允许发送。TCP_NODELAY 是取消 TCP 的确认延迟机制，相当于禁用了 Negale 算法。正常情况下，当 Server 端收到数据之后，它并不会马上向 client 端发送 ACK，而是会将 ACK 的发送延迟一段时间(一般是 40ms)，它希望在 t 时间内 server 端会向 client 端发送应答数据，这样 ACK 就能够和应答数据一起发送，就像是应答数据捎带着 ACK 过去。当然，TCP 确认延迟 40ms 并不是一直不变的， TCP 连接的延迟确认时间一般初始化为最小值 40ms，随后根据连接的重传超时时间(RTO)、上次收到数据包与本次接收数据包的时间间隔等参数进行不断调整。另外可以通过设置 TCP_QUICKACK 选项来取消确认延迟；未设置 TCP_CORK 选项时，若所有发出去的小数据包(包长度小于MSS)均被确认，则允许发送；上述条件都未满足，但发生了超时(一般为200ms)，则立即发送。

基于以上问题，TCP层肯定是会出现当次接收到的数据是不完整数据的情况。出现粘包可能的原因有：

发送方每次写入数据 < 套接字缓冲区大小；接收方读取套接字缓冲区数据不够及时。

出现半包的可能原因有：

发送方每次写入数据 > 套接字缓冲区大小；发送的数据大于协议 MTU，所以必须要拆包。

解决问题肯定不是在4层来做而是在应用层，通过定义通信协议来解决粘包和拆包的问题。发送方和接收方约定某个规则：

当发生粘包的时候通过某种约定来拆包；如果在拆包，通过某种约定来将数据组成一个完整的包处理。 2. 业界常用解决方案 1. 定长协议

指定一个报文具有固定长度。比如约定一个报文的长度是 5 字节，那么：

报文：1234，只有4字节，但是还差一个怎么办呢，不足部分用空格补齐。就变为：1234 。

如果不补齐空格，那么就会读到下一个报文的字节来填充上一个报文直到补齐为止，这样粘包了。

定长协议的优点是使用简单，缺点很明显：浪费带宽。

Netty 中提供了 FixedLengthFrameDecoder ，支持把固定的长度的字节数当做一个完整的消息进行解码。

2. 特殊字符分割协议

很好理解，在每一个你认为是一个完整的包的尾部添加指定的特殊字符，比如：\n，\r等等。

需要注意的是：约定的特殊字符要保证唯一性，不能出现在报文的正文中，否则就将正文一分为二了。

Netty 中提供了 DelimiterBasedFrameDecoder 根据特殊字符进行解码，LineBasedFrameDecoder默认以换行符作为分隔符。

3. 变长协议

变长协议的核心就是：将消息分为消息头和消息体，消息头中标识当前完整的消息体长度。

发送方在发送数据之前先获取数据的二进制字节大小，然后在消息体前面添加消息大小；接收方在解析消息时先获取消息大小，之后必须读到该大小的字节数才认为是完整的消息。

Netty 中提供了 LengthFieldBasedFrameDecoder ，通过 LengthFieldPrepender 来给实际的消息体添加 length 字段。

3. Netty 粘包演示

代码示例请看：github点我。

1. 实验主要逻辑

演示客户端发送多条消息，使用 Netty 自定义的 ByteBuf 作为传输数据格式，看看服务端接收数据是否是按每次发送的条数来接收还是按照当前缓冲区大小来接收。

主要代码：

Server:

package com.rickiyang.learn.packageEvent1; import io.netty.bootstrap.ServerBootstrap; import io.netty.channel.ChannelFuture; import io.netty.channel.EventLoopGroup; import io.netty.channel.nio.NioEventLoopGroup; import io.netty.channel.socket.nio.NioServerSocketChannel; import lombok.extern.slf4j.Slf4j; /** * @author: rickiyang * @date: 2020/3/15 * @description: server 端 */ @Slf4j public class PeServer { private int port; public PeServer(int port) { this.port = port; } public void start(){ EventLoopGroup bossGroup = new NioEventLoopGroup(); EventLoopGroup workGroup = new NioEventLoopGroup(); ServerBootstrap server = new ServerBootstrap().group(bossGroup,workGroup) .channel(NioServerSocketChannel.class) .childHandler(new ServerChannelInitializer()); try { ChannelFuture future = server.bind(port).sync(); future.channel().closeFuture().sync(); } catch (InterruptedException e) { log.error("server start fail",e); }finally { bossGroup.shutdownGracefully(); workGroup.shutdownGracefully(); } } public static void main(String[] args) { PeServer server = new PeServer(7788); server.start(); } }

ServerInitialzr:

package com.rickiyang.learn.packageEvent1; import io.netty.channel.ChannelInitializer; import io.netty.channel.ChannelPipeline; import io.netty.channel.socket.SocketChannel; /** * @author: rickiyang * @date: 2020/3/15 * @description: */ public class ServerChannelInitializer extends ChannelInitializer { @Override protected void initChannel(SocketChannel socketChannel) throws Exception { ChannelPipeline pipeline = socketChannel.pipeline(); // 自己的逻辑Handler pipeline.addLast("handler", new PeServerHandler()); } }

ServerHandler:

package com.rickiyang.learn.packageEvent1; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.SimpleChannelInboundHandler; import lombok.extern.slf4j.Slf4j; import java.nio.charset.StandardCharsets; /** * @author: rickiyang * @date: 2020/3/15 * @description: */ @Slf4j public class PeServerHandler extends SimpleChannelInboundHandler { private int counter; @Override public void channelActive(ChannelHandlerContext ctx) throws Exception { log.info("server channelActive"); } @Override protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception { ByteBuf buf = (ByteBuf) msg; byte[] req = new byte[buf.readableBytes()]; buf.readBytes(req); String body = new String(req, StandardCharsets.UTF_8); System.out.println("-----start------\n"+ body + "\n------end------"); String content = "receive" + ++counter; ByteBuf resp = Unpooled.copiedBuffer(content.getBytes()); ctx.writeAndFlush(resp); } @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { super.exceptionCaught(ctx, cause); ctx.close(); } }

服务端的 handler 主要逻辑是接收客户端发送过来的数据，看看是否是一条一条接收。然后每次接收到数据之后给客户端回复一个确认消息。

Client:

package com.rickiyang.learn.packageEvent1; import io.netty.bootstrap.Bootstrap; import io.netty.channel.ChannelFuture; import io.netty.channel.ChannelOption; import io.netty.channel.EventLoopGroup; import io.netty.channel.nio.NioEventLoopGroup; import io.netty.channel.socket.nio.NioSocketChannel; import lombok.extern.slf4j.Slf4j; /** * @author: rickiyang * @date: 2020/3/15 * @description: */ @Slf4j public class PeClient { private int port; private String address; public PeClient(int port, String address) { this.port = port; this.address = address; } public void start(){ EventLoopGroup group = new NioEventLoopGroup(); Bootstrap bootstrap = new Bootstrap(); bootstrap.group(group) .channel(NioSocketChannel.class) .option(ChannelOption.TCP_NODELAY, true) .handler(new ClientChannelInitializer()); try { ChannelFuture future = bootstrap.connect(address,port).sync(); future.channel().writeAndFlush("Hello world, i'm online"); future.channel().closeFuture().sync(); } catch (Exception e) { log.error("client start fail",e); }finally { group.shutdownGracefully(); } } public static void main(String[] args) { PeClient client = new PeClient(7788,"127.0.0.1"); client.start(); } }

ClientInitializer:

package com.rickiyang.learn.packageEvent1; import io.netty.channel.ChannelInitializer; import io.netty.channel.ChannelPipeline; import io.netty.channel.socket.SocketChannel; public class ClientChannelInitializer extends ChannelInitializer { @Override protected void initChannel(SocketChannel socketChannel) throws Exception { ChannelPipeline pipeline = socketChannel.pipeline(); // 客户端的逻辑 pipeline.addLast("handler", new PeClientHandler()); } }

ClientHandler:

package com.rickiyang.learn.packageEvent1; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.SimpleChannelInboundHandler; import lombok.extern.slf4j.Slf4j; import java.nio.charset.StandardCharsets; /** * @author: rickiyang * @date: 2020/3/15 * @description: */ @Slf4j public class PeClientHandler extends SimpleChannelInboundHandler { private int counter; @Override protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception { ByteBuf buf = (ByteBuf) msg; byte[] req = new byte[buf.readableBytes()]; buf.readBytes(req); String body = new String(req, StandardCharsets.UTF_8); System.out.println(body + " count:" + ++counter + "----end----\n"); } @Override public void channelActive(ChannelHandlerContext ctx) throws Exception { log.info("client channelActive"); byte[] req = ("我是一条测试消息，快来读我吧，啦啦啦").getBytes(); for (int i = 0; i < 100; i++) { ByteBuf message = Unpooled.buffer(req.length); message.writeBytes(req); ctx.writeAndFlush(message); } } @Override public void channelInactive(ChannelHandlerContext ctx) throws Exception { log.info("Client is close"); } }

客户端 handler 主要逻辑是：循环100次给服务端发送测试消息。接收服务端的确认消息。

启动项目之后我们来看看客户端和服务端分别收到的消息结果：

服务端接收到的消息：

-----start------ 我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，啦啦� ------end------ -----start------ ��我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，啦� ------end------ -----start------ �啦我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，啦 ------end------ -----start------ 啦啦我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，� ------end------ -----start------ ��啦啦我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧� ------end------ -----start------ �啦啦啦我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，啦啦啦我是一条测试消息，快来读我吧，啦啦啦 ------end------

这里能看到多条消息被粘到一起发送了。

客户端接收到服务端回传的消息：

receive1receive2receive3receive4receive5 count:1----end---- receive6 count:2----end----

服务端收到 6 次消息，所以回复了 6 次，同样客户端接收消息也出现粘包的现象。

因为我们并没有对数据包做任何声明，站在 TCP 协议端看， Netty 属于应用层，我们上面的示例代码中未对原始的数据包做任何处理。

4. Netty 粘包处理

处理 TCP 粘包的唯一方法就是制定应用层的数据通讯协议，通过协议来规范现有接收的数据是否满足消息数据的需要。

1. Netty 提供的能力

为了解决网络数据流的拆包粘包问题，Netty 为我们内置了如下的解码器：

ByteToMessageDecoder：如果想实现自己的半包解码器，实现该类； MessageToMessageDecoder：一般作为二次解码器，当我们在 ByteToMessageDecoder 将一个 bytes 数组转换成一个 java 对象的时候，我们可能还需要将这个对象进行二次解码成其他对象，我们就可以继承这个类； LineBasedFrameDecoder：通过在包尾添加回车换行符 \r\n 来区分整包消息； StringDecoder：字符串解码器； DelimiterBasedFrameDecoder：特殊字符作为分隔符来区分整包消息； FixedLengthFrameDecoder：报文大小固定长度，不够空格补全； ProtoBufVarint32FrameDecoder：通过 Protobuf 解码器来区分整包消息； ProtobufDecoder： Protobuf 解码器； LengthFieldBasedFrameDecoder：指定长度来标识整包消息，通过在包头指定整包长度来约定包长。

Netty 还内置了如下的编码器：

ProtobufEncoder：Protobuf 编码器； MessageToByteEncoder：将 Java 对象编码成 ByteBuf； MessageToMessageEncoder：如果不想将 Java 对象编码成 ByteBuf，而是自定义类就继承这个； LengthFieldPrepender：LengthFieldPrepender 是一个非常实用的工具类，如果我们在发送消息的时候采用的是：消息长度字段+原始消息的形式，那么我们就可以使用 LengthFieldPrepender。这是因为 LengthFieldPrepender 可以将待发送消息的长度（二进制字节长度）写到 ByteBuf 的前两个字节。

编解码相关类结构图如下：

上面的类关系能看到所有的自定义解码器都是继承自 ByteToMessageDecoder。在Netty 中 Decoder 主要分为两大类：

一种是将字节流转换为某种协议的数据格式：ByteToMessageDecoder和ReplayingDecoder；一种是将一直协议的数据转为另一种协议的数据格式：MessageToMessageDecoder。

将字节流转为对象是一种很常见的操作，也是一个消息框架应该提供的基础功能。因为 Decoder 的作用是将输入的数据解析成特定协议，上图中可以看到所有的 Decoder 都实现了 ChannelInboundHandler接口。在应用层将 byte 转为 message 的难度在于如何确定当前的包是一个完整的数据包，有两种方案可以实现：

监听当前 socket 的线程一直等待，直到收到的 byte 可以完成的构成一个包为止。这种方式的弊端就在于要浪费一个线程去等。第二种方案是为每个监听的 socket 都构建一个本地缓存，当前监听线程如果遇到字节数不够的情况就先将获取到的数据存入缓存，继而处理别的请求，等到这里有数据的时候再来将新数据继续写入缓存直到数据构成一个完整的包取出。

ByteToMessageDecoder 采用的是第二种方案。在 ByteToMessageDecoder 中有一个对象 ByteBuf，该对象用于存储当前 Decoder接收到的 byte 数据。

public abstract class ByteToMessageDecoder extends ChannelInboundHandlerAdapter { // 用来保存累计读取到的字节. 我们读到的新字节会保存(缓冲)在这里 ByteBuf cumulation; // 用来做累计的，负责将读到的新字节写入 cumulation，有两个实现 MERGE_CUMULATOR 和 COMPOSITE_CUMULATOR private Cumulator cumulator = MERGE_CUMULATOR; //设置为true后, 单个解码器只会解码出一个结果 private boolean singleDecode; private boolean decodeWasNull; //是否是第一次读取数据 private boolean first; //多少次读取后, 丢弃数据默认16次 private int discardAfterReads = 16; //已经累加了多少次数据 private int numReads; //每次接收到数据，就会调用channelRead 进行处理 //该处理器用于处理二进制数据，所以 msg 字段的类型应该是 ByteBuf。 //如果不是，则交给pipeLine的下一个处理器进行处理。 //下面的代码中可以看出 @Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { //如果不是ByteBuf则不处理 if (msg instanceof ByteBuf) { //out用于存储解析二进制流得到的结果，一个二进制流可能会解析出多个消息，所以out是一个list CodecOutputList out = CodecOutputList.newInstance(); try { ByteBuf data = (ByteBuf) msg; //判断cumulation == null;并将结果赋值给first。因此如果first为true，则表示第一次接受到数据 first = cumulation == null; //如果是第一次接受到数据，直接将接受到的数据赋值给缓存对象cumulation if (first) { cumulation = data; } else { // 第二次解码，就将 data 向 cumulation 追加，并释放 data //如果cumulation中的剩余空间，不足以存储接收到的data，将cumulation扩容 cumulation = cumulator.cumulate(ctx.alloc(), cumulation, data); } // 得到追加后的 cumulation 后，调用 decode 方法进行解码 // 解码过程中，调用 fireChannelRead 方法，主要目的是将累积区的内容 decode 到数组中 callDecode(ctx, cumulation, out); } catch (DecoderException e) { throw e; } catch (Throwable t) { throw new DecoderException(t); } finally { //如果cumulation没有数据可读了，说明所有的二进制数据都被解析过了 //此时对cumulation进行释放，以节省内存空间。 //反之cumulation还有数据可读，那么if中的语句不会运行，因为不对cumulation进行释放 //因此也就缓存了用户尚未解析的二进制数据。 if (cumulation != null && !cumulation.isReadable()) { // 将次数归零 numReads = 0; // 释放累计区 cumulation.release(); // 等待 gc cumulation = null; // 如果超过了 16 次，就压缩累计区，主要是将已经读过的数据丢弃，将 readIndex 归零。 } else if (++ numReads >= discardAfterReads) { // We did enough reads already try to discard some bytes so we not risk to see a OOME. // See https://github.com/netty/netty/issues/4275 numReads = 0; discardSomeReadBytes(); } int size = out.size(); // 如果没有向数组插入过任何数据 decodeWasNull = !out.insertSinceRecycled(); // 循环数组，向后面的 handler 发送数据，如果数组是空，那不会调用 fireChannelRead(ctx, out, size); // 将数组中的内容清空，将数组的数组的下标恢复至原来 out.recycle(); } } else { //如果msg类型是不是ByteBuf，直接调用下一个handler进行处理 ctx.fireChannelRead(msg); } } //callDecode方法主要用于解析cumulation 中的数据，并将解析的结果放入List out中。 //由于cumulation中缓存的二进制数据，可能包含了出多条有效信息，因此在callDecode方法中，默认会调用多次decode方法 //我们在覆写decode方法时，每次只解析一个消息，添加到out中，callDecode通过多次回调decode //每次传递进来都是相同的List out实例，因此每一次解析出来的消息，都存储在同一个out实例中。 //当cumulation没有数据可以继续读，或者某次调用decode方法后，List out中元素个数没有变化，则停止回调decode方法。 protected void callDecode(ChannelHandlerContext ctx, ByteBuf in, List out) { try { //如果cumulation中有数据可读的话，一直循环调用decode while (in.isReadable()) { //获取上一次decode方法调用后，out中元素数量，如果是第一次调用，则为0。 int outSize = out.size(); //上次循环成功解码 if (outSize > 0) { //用后面的业务 handler 的 ChannelRead 方法读取解析的数据 fireChannelRead(ctx, out, outSize); out.clear(); if (ctx.isRemoved()) { break; } outSize = 0; } int oldInputLength = in.readableBytes(); //回调decode方法，由开发者覆写，用于解析in中包含的二进制数据，并将解析结果放到out中。 decode(ctx, in, out); if (ctx.isRemoved()) { break; } //outSize是上一次decode方法调用时out的大小，out.size()是当前out大小 //如果二者相等，则说明当前decode方法调用没有解析出有效信息。 if (outSize == out.size()) { //此时，如果发现上次decode方法和本次decode方法调用候，in中的剩余可读字节数相同 //则说明本次decode方法没有读取任何数据解析 //(可能是遇到半包等问题，即剩余的二进制数据不足以构成一条消息)，跳出while循环。 if (oldInputLength == in.readableBytes()) { break; } else { continue; } } //处理人为失误。如果走到这段代码，则说明outSize != out.size()。 //也就是本次decode方法实际上是解析出来了有效信息放到out中。 //但是oldInputLength == in.readableBytes()，说明本次decode方法调用并没有读取任何数据 //但是out中元素却添加了。 //这可能是因为开发者错误的编写了代码，例如mock了一个消息放到List中。 if (oldInputLength == in.readableBytes()) { throw new DecoderException( StringUtil.simpleClassName(getClass()) + ".decode() did not read anything but decoded a message."); } if (isSingleDecode()) { break; } } } catch (DecoderException e) { throw e; } catch (Throwable cause) { throw new DecoderException(cause); } } }

这里 channelRead()的主要逻辑是：

从对象池中取出一个空的数组；判断成员变量是否是第一次使用，要注意的是，这里既然使用了成员变量，所以这个 handler 不能是 @Shareble 状态的 handler，不然你就分不清成员变量是哪个 channel 的。将 unsafe 中传递来的数据写入到这个 cumulation 累积区中；写到累积区后，调用子类的 decode 方法，尝试将累积区的内容解码，每成功解码一个，就调用后面节点的 channelRead 方法。若没有解码成功，什么都不做；如果累积区没有未读数据了，就释放累积区；如果还有未读数据，且解码超过了 16 次（默认），就对累积区进行压缩。将读取过的数据清空，也就是将 readIndex 设置为0；设置 decodeWasNull 的值，如果上一次没有插入任何数据，这个值就是 ture。该值在调用 channelReadComplete 方法的时候，会触发 read 方法（不是自动读取的话），尝试从 JDK 的通道中读取数据，并将之前的逻辑重来。主要应该是怕如果什么数据都没有插入，就执行 channelReadComplete 会遗漏数据；调用 fireChannelRead 方法，将数组中的元素发送到后面的 handler 中；将数组清空。并还给对象池。

当数据添加到累积区之后，需要调用 decode 方法进行解码，代码见上面的 callDecode()方法。在 callDecode()中最关键的代码就是将解析完的数据拿取调用decode(ctx, in, out)方法。所以如果继承 ByteToMessageDecoder 类实现自己的字节流转对象的逻辑我们就要覆写该方法。

2. LineBasedFrameDecoder 使用

LineBasedFrameDecoder 通过在包尾添加回车换行符 \r\n 来区分整包消息。逻辑比较简单，示例代码见：

示例代码见：LineBasedFrameDecoder gitHub示例

3. FixedLengthFrameDecoder 使用

LineBasedFrameDecoder即固定消息长度解码器，个人认为这个貌似不能适用通用场景。

示例代码见：FixedLengthFrameDecoder gitHub 示例

4. DelimiterBasedFrameDecoder 使用

DelimiterBasedFrameDecoder即自定义分隔符解码器。相当于是 LineBasedFrameDecoder的高阶版。

示例代码见：DelimiterBasedFrameDecoder gitHub示例

5. LengthFieldBasedFrameDecoder 使用

LengthFieldBasedFrameDecoder相对就高端一点。前面我们使用到的拆包都是基于一些约定来做的，比如固定长度，特殊分隔符，这些方案总是有一定的弊端。最好的方案就是：发送方告诉我当前消息总长度，接收方如果没有收到该长度大小的数据就认为是没有收完继续等待。

先看一下该类的构造函数：

/** * Creates a new instance. * * @param maxFrameLength 帧的最大长度 * * @param lengthFieldOffset 长度字段偏移的地址 * * @param lengthFieldLength 长度字段所占的字节长 * 修改帧数据长度字段中定义的值，可以为负数因为有时候我们习惯把头部记入长度, * 若为负数,则说明要推后多少个字段 * @param lengthAdjustment 解析时候跳过多少个长度 * * @param initialBytesToStrip 解码出一个数据包之后，去掉开头的字节数 * * @param initialBytesToStrip 为true，当frame长度超过maxFrameLength时立即报 * TooLongFrameException异常，为false，读取完整个帧再报异 * */ public LengthFieldBasedFrameDecoder( int maxFrameLength, int lengthFieldOffset, int lengthFieldLength, int lengthAdjustment, int initialBytesToStrip) { this( maxFrameLength, lengthFieldOffset, lengthFieldLength, lengthAdjustment, initialBytesToStrip, true); }

在 LengthFieldBasedFrameDecoder类的注解上给出了一些关于该类使用的示例：

示例1：

lengthFieldOffset = 0，长度字段偏移位置为0表示从包的第一个字节开始读取；

lengthFieldLength = 2，长度字段长为2，从包的开始位置往后2个字节的长度为长度字段；

lengthAdjustment = 0 ，解析的时候无需跳过任何长度；

initialBytesToStrip = 0，无需去掉当前数据包的开头字节数, header + body。

0x000C 转为 int = 12。

* * lengthFieldOffset = 0 * lengthFieldLength = 2 * lengthAdjustment = 0 * initialBytesToStrip = 0 (= do not strip header) * * BEFORE DECODE (14 bytes) AFTER DECODE (14 bytes) * +--------+----------------+ +--------+----------------+ * | Length | Actual Content |----->| Length | Actual Content | * | 0x000C | "HELLO, WORLD" | | 0x000C | "HELLO, WORLD" | * +--------+----------------+ +--------+----------------+ *

上面这个设置表示：body长度为12，从当前包的第0个字节开始读取，前两个字节表示包长度，读取数据 body的时候不偏移从0字节开始，所以整包大小14个字节，包含包头长度字节在内。

示例2：

lengthFieldOffset = 0，长度字段偏移位置为0表示从包的第一个字节开始读取；

lengthFieldLength = 2，长度字段长为2，从包的开始位置往后2个字节的长度为长度字段；

lengthAdjustment = 0 ，解析的时候无需跳过任何长度；

initialBytesToStrip = 2，去掉当前数据包的开头2字节，去掉 header。

0x000C 转为 int = 12。

这个配置跟上面的而区别就在于，initialBytesToStrip = 2，表示当前包中的有效数据是从整包偏移2个字节开始计算的，即包头中的长度字段 2 byte 不属于包内容的一部分。

示例3：

lengthFieldOffset = 0，长度字段偏移位置为0表示从包的第一个字节开始读取；

lengthFieldLength = 2，长度字段长为2，从包的开始位置往后2个字节的长度为长度字段；

lengthAdjustment = -2 ，解析的时候无需跳过任何长度；

initialBytesToStrip = 0，无需去掉当前数据包的开头字节数。

0x000C 转为 int = 12。

* * lengthFieldOffset = 0 * lengthFieldLength = 2 * lengthAdjustment = -2 (= the length of the Length field) * initialBytesToStrip = 0 * * BEFORE DECODE (14 bytes) AFTER DECODE (14 bytes) * +--------+----------------+ +--------+----------------+ * | Length | Actual Content |----->| Length | Actual Content | * | 0x000E | "HELLO, WORLD" | | 0x000E | "HELLO, WORLD" | * +--------+----------------+ +--------+----------------+ *

length = 14，长度字段为 2 字节，真实的数据长度为 12 个字节，但是 length = 14，那么说明 length的长度也算上了数据包长度了。lengthAdjustment = -2 ，表示当前length长度往回调2个字节，这样总包长度就是14个字节。

示例4：

lengthFieldOffset = 2，长度字段偏移位置为2表示从包的第3个字节开始读取；

lengthFieldLength = 3，长度字段长为3，从包的开始位置往后3个字节的长度为长度字段；

lengthAdjustment = 0 ，解析的时候无需跳过任何长度；

initialBytesToStrip = 0，无需去掉当前数据包的开头字节数。

0x000E 转为 int = 14。

header头占2个字节，长度字段占3个字节，content字段占12个字节，总共17个字节。body读取无偏移要求，所以body整体也是17个字节。

示例5：

lengthFieldOffset = 0，长度字段偏移位置为0表示从包的第0个字节开始读取；

lengthFieldLength = 3，长度字段长为3，从包的开始位置往后3个字节的长度为长度字段；

lengthAdjustment = 2 ，解析的时候跳过2个字节；

initialBytesToStrip = 0，无需去掉当前数据包的开头字节数。

0x000C 转为 int = 12。

这个包 length在最前面传输占3个字节，header在中间占两个字节，content在最后占12个字节。body字段只有content，所以读取content的时候需要在length字段的基础上往前偏移2个字节跳过heade字段。

关于 LengthFieldBasedFrameDecoder 构造函数的示例用法我们先将这么多，下来举一个示例我们看看实际中的使用：

示例代码见：LengthFieldBasedFrameDecoder基本使用 gitHub示例

代码解释：

@Slf4j public class PeClientHandler extends SimpleChannelInboundHandler { @Override public void channelActive(ChannelHandlerContext ctx) throws Exception { log.info("client channelActive"); for (int i = 0; i < 100; i++) { byte[] req = ("我是一条测试消息，快来读我吧，啦啦啦" + i).getBytes(); ByteBuf message = Unpooled.buffer(req.length); message.writeInt(req.length); message.writeBytes(req); ctx.writeAndFlush(message); } } }

客户端发送消息是：int型的length字段占4个字节，剩余字节为content内容。那么对应到客户端接收的解码器设置：

pipeline.addLast(new LengthFieldBasedFrameDecoder(1024, // 帧的最大长度，即每个数据包最大限度 0, // 长度字段偏移量 4, // 长度字段所占的字节数 0, // 消息头的长度，可以为负数 4) // 需要忽略的字节数，从消息头开始，这里是指整个包 );

长度字段4个字节，消息体忽略4字节，即排除长度字段之后的内容算是body。

以上的这段演示代码的重点，大家可以下载示例功能，自己演示一下。

但是有个问题是：我们上面写的示例代码在生产环境中只能是玩具。消息体的读取配置不应该在这里通过参数配置来设置，应该有一个约定的消息结构体，每一个字段是什么数据结构会占用多大空间都应该在结构体中约定清楚。每个字段读取对应空间大小的数据剩下的就是别人的部分互不侵犯。

所以下面的一个示例给出了通过继承 LengthFieldBasedFrameDecoder 重写 decode 方法来实现解析出约定对象的实现。

6. 自定义编解码器的 LengthFieldBasedFrameDecoder 使用

首先我们自定义了一个消息体：

public class MsgReq { private byte type; private int length; private String content; }

包含3个字段。

发送消息出去的时候肯定是要将对象转为 byte 发送，所以需要一个消息编码器，我们继承 MessageToByteEncoder 来实现编码器：

package com.rickiyang.learn.packageEvent5; import io.netty.buffer.ByteBuf; import io.netty.channel.ChannelHandlerContext; import io.netty.handler.codec.MessageToByteEncoder; import java.nio.charset.StandardCharsets; /** * @author rickiyang * @date 2020-05-14 * @Desc 自定义编码器 */ public class MyProtocolEncoder extends MessageToByteEncoder { @Override protected void encode(ChannelHandlerContext ctx, Object msg, ByteBuf out) throws Exception { MsgReq req = (MsgReq) msg; out.writeByte(req.getType()); out.writeInt(req.getLength()); out.writeBytes(req.getContent().getBytes(StandardCharsets.UTF_8)); } }

即将 MsgReq 对象转为对应的 byte 发送。

发送出去的是 byte 字节，对应的解码器应该是将 byte 转为对象。自然解码器应该是继承 ByteToMessageDecoder。我们的目的不是自己实现一个完完全全的自定义解码器，而是在消息长度解码器的基础上完成对象解析的工作，所以解码器如下：

package com.rickiyang.learn.packageEvent5; import io.netty.buffer.ByteBuf; import io.netty.channel.ChannelHandlerContext; import io.netty.handler.codec.LengthFieldBasedFrameDecoder; import java.nio.charset.StandardCharsets; /** * @author rickiyang * @date 2020-05-14 * @Desc 自定义解码器 */ public class MyProtocolDecoder extends LengthFieldBasedFrameDecoder { /** * @param maxFrameLength 帧的最大长度 * @param lengthFieldOffset length字段偏移的地址 * @param lengthFieldLength length字段所占的字节长 * @param lengthAdjustment 修改帧数据长度字段中定义的值，可以为负数因为有时候我们习惯把头部记入长度,若为负数,则说明要推后多少个字段 * @param initialBytesToStrip 解析时候跳过多少个长度 * @param failFast 为true，当frame长度超过maxFrameLength时立即报TooLongFrameException异常，为false，读取完整个帧再报异 */ public MyProtocolDecoder(int maxFrameLength, int lengthFieldOffset, int lengthFieldLength, int lengthAdjustment, int initialBytesToStrip, boolean failFast) { super(maxFrameLength, lengthFieldOffset, lengthFieldLength, lengthAdjustment, initialBytesToStrip, failFast); } @Override protected Object decode(ChannelHandlerContext ctx, ByteBuf in) throws Exception { //在这里调用父类的方法 in = (ByteBuf) super.decode(ctx, in); if (in == null) { return null; } //读取type字段 byte type = in.readByte(); //读取length字段 int length = in.readInt(); if (in.readableBytes() != length) { throw new RuntimeException("长度与标记不符"); } //读取body byte[] bytes = new byte[in.readableBytes()]; in.readBytes(bytes); return MsgReq.builder().length(length).type(type).content(new String(bytes, StandardCharsets.UTF_8)).build(); } }

通过这种方式，我们只用约定好消息的最大长度，比如一条消息超过多少字节就拒收，约定好消息长度字段所占的字节，一般来说int类型4个字节足够。剩下的几个参数都无需设置，按照约定的消息格式进行解析即可。

示例代码见：LengthFieldBasedFrameDecoder自定义编解码器 gitHub示例

5. 小结

本篇将了关于 Netty 中处理拆包粘包的一些实用工具以及如果实现自定义的编解码器的方式。每种处理方式都给出了对应的案例操作，大家有兴趣的可以下载代码自行运行看看处理效果。后面也给出了关于自定义编解码器的示例，大家如果有兴趣可以自己写一下编解码操作，下一篇再一起看看编解码器在消息读写过程被使用在哪个阶段。

【本文地址】

Netty 中的粘包和拆包

Netty 中的粘包和拆包

今日新闻

推荐新闻